package org.llc.flink.stream.mysql;

import com.google.gson.Gson;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FlinkJdbcClient {


    public static void main(String[] args) throws Exception {

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Tuple4<String, String, Integer, String>> dataStream = env
                .addSource(new Flink2JdbcSource());

        // tranfomat
        DataStream resultStream = dataStream.map((MapFunction<Tuple4<String, String, Integer, String>, Object>) d -> {
            Tuple4<String, String, Integer, String> result = new Tuple4<>();
            result.f0 = d.f0.toUpperCase();
            result.f1 = d.f1;
            result.f2 = d.f2;
            result.f3 = d.f3;
            System.out.println(new Gson().toJson(result));
            return result;
        });
        resultStream.addSink(new Flink2JdbcSink());
        env.execute("Flink cost DB data to write Database");

    }
}